package com.yl.flink.processor;

import cn.hutool.core.date.DateUtil;
import com.yl.constant.Warn;
import com.yl.entity.MultiDataEntity;
import com.yl.entity.PrimaryWarn;
import com.yl.entity.cdc.SettingsDataQuota;
import com.yl.entity.cdc.SettingsThreshold;
import org.apache.flink.api.common.functions.FlatMapFunction;
import org.apache.flink.util.Collector;

import java.util.Comparator;
import java.util.List;
import java.util.Map;
import java.util.Optional;

/**
 * @author wlf
 * @since 2022/8/19
 */
public class WarnJudgeFunc implements FlatMapFunction<MultiDataEntity, PrimaryWarn> {

    /**
     * 判断每个指标是否有告警产生，每个指标最终只返回最高等级的告警
     *
     * @param multiDataEntity 流元素
     * @param out             输出器
     */
    @Override
    public void flatMap(MultiDataEntity multiDataEntity, Collector<PrimaryWarn> out) throws Exception {
        Map<Integer, Double> rawValues = multiDataEntity.getRawValues();
        // 各指标对应的各等级阈值信息
        Map<Integer, List<SettingsThreshold>> ths = multiDataEntity.getThs();
        // 判断是原始流元素还是计算流元素
        Optional
                .ofNullable(multiDataEntity.getCalValues())
                .ifPresentOrElse(calValues -> {
                    // 计算流元素
                    Integer quotaCode = calValues.f0;
                    Double quotaValue = calValues.f1;
                    rawValues.put(quotaCode, quotaValue);
                    Optional
                            .ofNullable(ths.get(quotaCode))
                            .flatMap(thresholds -> Optional
                                    .ofNullable(doJudge(thresholds, quotaValue, quotaCode, multiDataEntity)))
                            .ifPresent(out::collect);
                }, () -> {
                    // 原始流元素
                    for (Integer quotaCode : ths.keySet()) {
                        Double quotaValue = rawValues.get(quotaCode);
                        Optional
                                .ofNullable(ths.get(quotaCode))
                                .flatMap(thresholds -> Optional
                                        .ofNullable(doJudge(thresholds, quotaValue, quotaCode, multiDataEntity)))
                                .ifPresent(out::collect);
                    }
                });
    }

    /**
     * 告警判断
     */
    private PrimaryWarn doJudge(List<SettingsThreshold> thresholds, Double quotaValue, int quotaCode, MultiDataEntity multiDataEntity) {
        double value = quotaValue;
        // 按照等级排序
        thresholds.sort(Comparator.comparingInt(SettingsThreshold::getLevel));
        for (SettingsThreshold th : thresholds) {
            Integer level = th.getLevel();
            Double lower = th.getLower();
            Double upper = th.getUpper();
            // 是否取绝对值(1是，0否）
            Integer absolute = th.getAbsolute();
            if (absolute == 1) {
                value = Math.abs(value);
            }
            if (value >= lower && value < upper) {
                // 获取指标信息
                SettingsDataQuota quota = null;
                for (SettingsDataQuota settingsDataQuota : multiDataEntity.getQuotas()) {
                    if (settingsDataQuota.getCode() == quotaCode) {
                        quota = settingsDataQuota;
                    }
                }
                // 每个指标只产生一个最高等级的告警
                return PrimaryWarn.builder()
                        .multiDataEntity(multiDataEntity)
                        .quota(quota)
                        .th(th)
                        .warn(Warn.getWarn(level))
                        .warnTime(DateUtil.date(multiDataEntity.getPickTime() / 1000))
                        .dataValue(quotaValue)
                        .message(Warn.msg(level))
                        .time(DateUtil.date())
                        .build();
            }
        }
        return null;
    }

}
